块-将输入发送到python子进程管道。-blocks-sendinputtopythonsubprocesspipeline

21  

I found out how to do it.

我知道怎么做了。

It is not about threads, and not about select().

它不是关于线程的，而不是关于select()的。

When I run the first process (grep), it creates two low-level file descriptors, one for each pipe. Lets call those a and b.

当我运行第一个进程(grep)时，它将创建两个低级的文件描述符，一个用于每个管道。我们称之为a和b。

When I run the second process, b gets passed to cut sdtin. But there is a brain-dead default on Popen - close_fds=False.

当我运行第二个进程时，b被传递给了sdtin。但是，在Popen - close_fds=False中存在一个脑死亡的默认值。

The effect of that is that cut also inherits a. So grep can't die even if I close a, because stdin is still open on cut's process (cut ignores it).

它的作用是，切割也继承了a，所以即使我关闭a, grep也不会死，因为stdin在cut的过程中仍然是开放的(cut忽略它)。

The following code now runs perfectly.

下面的代码现在运行得很好。

from subprocess import Popen, PIPE

p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE, close_fds=True)
p1.stdin.write('Hello World\n')
p1.stdin.close()
result = p2.stdout.read() 
assert result == "Hello Worl\n"

close_fds=True SHOULD BE THE DEFAULT on unix systems. On windows it closes all fds, so it prevents piping.

在unix系统中，close_fds=True应该是默认值。在windows上，它关闭所有的fds，因此它可以防止管道。

EDIT:

编辑:

PS: For people with a similar problem reading this answer: As pooryorick said in a comment, that also could block if data written to p1.stdin is bigger than the buffers. In that case you should chunk the data into smaller pieces, and use select.select() to know when to read/write. The code in the question should give a hint on how to implement that.

对于有类似问题的人来说，阅读这个答案:正如pooryorick在评论中所说的那样，如果数据写到p1，那也会阻塞。stdin比缓冲区大。在这种情况下，你应该把数据分成小块，并使用select.select()来知道什么时候读/写。问题中的代码应该给出如何实现这一点的提示。

EDIT2: Found another solution, with more help from pooryorick - instead of using close_fds=True and close ALL fds, one could close the fds that belongs to the first process, when executing the second, and it will work. The closing must be done in the child so the preexec_fn function from Popen comes very handy to do just that. On executing p2 you can do:

EDIT2:找到了另一个解决方案，从pooryorick那里得到了更多的帮助——而不是使用close_fds=True并关闭所有fds，一个可以关闭属于第一个进程的fds，当执行第二个进程时，它将工作。关闭必须在child中完成，所以Popen的preexec_fn函数非常方便。在执行p2时，你可以:

p2 = Popen(cmd2, stdin=p1.stdout, stdout=PIPE, stderr=devnull, preexec_fn=p1.stdin.close)

6  

Working with large files

Two principles need to be applied uniformly when working with large files in Python.

在使用Python中的大文件时，需要一致地应用两个原则。

1. Since any IO routine can block, we must keep each stage of the pipeline in a different thread or process. We use threads in this example, but subprocesses would let you avoid the GIL.
2. 由于任何IO例程都可以阻塞，我们必须在不同的线程或进程中保持管道的每个阶段。我们在这个示例中使用线程，但是子进程会让您避免GIL。
3. We must use incremental reads and writes so that we don't wait for EOF before starting to make progress.
4. 我们必须使用增量读和写，这样我们才不会等待EOF才开始取得进展。

An alternative is to use nonblocking IO, though this is cumbersome in standard Python. See gevent for a lightweight threading library that implements the synchronous IO API using nonblocking primitives.

另一种方法是使用非阻塞IO，尽管这在标准Python中很麻烦。请参阅gevent，以实现使用非阻塞原语实现同步IO API的轻量级线程库。

Example code

We'll construct a silly pipeline that is roughly

我们将构建一个简单的管道。

{cat /usr/share/dict/words} | grep -v not              \
    | {upcase, filtered tee to stderr} | cut -c 1-10   \
    | {translate 'E' to '3'} | grep K | grep Z | {downcase}

where each stage in braces {} is implemented in Python while the others use standard external programs. TL;DR: See this gist.

在大括号{}中的每个阶段都是在Python中实现的，而其他阶段则使用标准的外部程序。TL;博士:看到这个要点。

We start with the expected imports.

我们从预期的进口开始。

#!/usr/bin/env python
from subprocess import Popen, PIPE
import sys, threading

Python stages of the pipeline

All but the last Python-implemented stage of the pipeline needs to go in a thread so that it's IO does not block the others. These could instead run in Python subprocesses if you wanted them to actually run in parallel (avoid the GIL).

除了最后一个python实现阶段之外，所有的线程都需要进入一个线程，这样它的IO就不会阻塞其他线程。如果您希望它们实际上并行运行(避免GIL)，那么它们可以在Python子进程中运行。

def writer(output):
    for line in open('/usr/share/dict/words'):
        output.write(line)
    output.close()
def filter(input, output):
    for line in input:
        if 'k' in line and 'z' in line: # Selective 'tee'
            sys.stderr.write('### ' + line)
        output.write(line.upper())
    output.close()
def leeter(input, output):
    for line in input:
        output.write(line.replace('E', '3'))
    output.close()

Each of these needs to be put in its own thread, which we'll do using this convenience function.

每一个都需要放到它自己的线程中，我们将使用这个便利函数。

def spawn(func, **kwargs):
    t = threading.Thread(target=func, kwargs=kwargs)
    t.start()
    return t

Create the pipeline

Create the external stages using Popen and the Python stages using spawn. The argument bufsize=-1 says to use the system default buffering (usually 4 kiB). This is generally faster than the default (unbuffered) or line buffering, but you'll want line buffering if you want to visually monitor the output without lags.

使用Popen和使用衍生工具的Python阶段创建外部阶段。参数bufsize=-1表示使用系统默认缓冲(通常为4 kiB)。这通常比默认的(未缓冲的)或行缓冲要快，但是如果您想在不滞后的情况下直观地监视输出，您将需要行缓冲。

grepv   = Popen(['grep','-v','not'], stdin=PIPE, stdout=PIPE, bufsize=-1)
cut     = Popen(['cut','-c','1-10'], stdin=PIPE, stdout=PIPE, bufsize=-1)
grepk = Popen(['grep', 'K'], stdin=PIPE, stdout=PIPE, bufsize=-1)
grepz = Popen(['grep', 'Z'], stdin=grepk.stdout, stdout=PIPE, bufsize=-1)

twriter = spawn(writer, output=grepv.stdin)
tfilter = spawn(filter, input=grepv.stdout, output=cut.stdin)
tleeter = spawn(leeter, input=cut.stdout, output=grepk.stdin)

Drive the pipeline

Assembled as above, all the buffers in the pipeline will fill up, but since nobody is reading from the end (grepz.stdout), they will all block. We could read the entire thing in one call to grepz.stdout.read(), but that would use a lot of memory for large files. Instead, we read incrementally.

如上所述，管道中的所有缓冲区都将被填满，但由于没有人从末尾读取(grepz.stdout)，它们将全部阻塞。我们可以在一个调用grepz.stdout.read()中读取整个事件，但是这会占用大量内存用于大型文件。相反,我们读增量。

for line in grepz.stdout:
    sys.stdout.write(line.lower())

The threads and processes clean up once they reach EOF. We can explicitly clean up using

线程和进程一旦到达EOF就会清除。我们可以明确地使用。

for t in [twriter, tfilter, tleeter]: t.join()
for p in [grepv, cut, grepk, grepz]: p.wait()

Python-2.6 and earlier

Internally, subprocess.Popen calls fork, configures the pipe file descriptors, and calls exec. The child process from fork has copies of all file descriptors in the parent process, and both copies will need to be closed before the corresponding reader will get EOF. This can be fixed by manually closing the pipes (either by close_fds=True or a suitable preexec_fn argument to subprocess.Popen) or by setting the FD_CLOEXEC flag to have exec automatically close the file descriptor. This flag is set automatically in Python-2.7 and later, see issue12786. We can get the Python-2.7 behavior in earlier versions of Python by calling

在内部,子流程。Popen调用fork，配置管道文件描述符，并调用exec。来自fork的子进程在父进程中具有所有文件描述符的副本，并且在相应的阅读器将得到EOF之前，两个副本都需要被关闭。这可以通过手动关闭管道(通过close_fds=True或一个合适的preexec_fn参数到subprocess.Popen)来修复，或者通过设置FD_CLOEXEC标志，使exec自动关闭文件描述符。此标志自动设置在Python-2.7中，稍后将看到问题12786。我们可以通过调用来获得Python早期版本中的Python-2.7行为。

p._set_cloexec_flags(p.stdin)

before passing p.stdin as an argument to a subsequent subprocess.Popen.

之前通过p。将stdin作为后续子进程的参数。

3  

There are three main tricks to making pipes work as expected

有三种方法可以让管道按照预期工作。

1. Make sure each end of the pipe is used in a different thread/process (some of the examples near the top suffer from this problem).

   确保管道的每一端都在不同的线程/进程中使用(在顶部附近的一些示例会受到这个问题的影响)。

2. explicitly close the unused end of the pipe in each process

   在每个进程中显式关闭管道未使用的端口。

3. deal with buffering by either disabling it (Python -u option), using pty's, or simply filling up the buffer with something that won't affect the data, ( maybe '\n', but whatever fits).

   通过禁用它(Python -u选项)，使用pty，或者简单地用一些不会影响数据的东西填充缓冲区(可能是“\n”，但无论如何)。

The examples in the Python "pipeline" module (I'm the author) fit your scenario exactly, and make the low-level steps fairly clear.

Python“管道”模块中的示例(我是作者)完全适合您的场景，并使底层步骤相当清晰。

http://pypi.python.org/pypi/pipeline/

http://pypi.python.org/pypi/pipeline/

More recently, I used the subprocess module as part of a producer-processor-consumer-controller pattern:

最近，我使用子流程模块作为生产者-处理器-消费者-控制器模式的一部分:

http://www.darkarchive.org/w/Pub/PythonInteract

http://www.darkarchive.org/w/Pub/PythonInteract

This example deals with buffered stdin without resorting to using a pty, and also illustrates which pipe ends should be closed where. I prefer processes to threading, but the principle is the same. Additionally, it illustrates synchronizing Queues to which feed the producer and collect output from the consumer, and how to shut them down cleanly (look out for the sentinels inserted into the queues). This pattern allows new input to be generated based on recent output, allowing for recursive discovery and processing.

这个示例处理的是缓冲的stdin，而不需要使用pty，还演示了应该在何处关闭管道端口。我更喜欢进程而不是线程，但原理是一样的。此外，它还演示了同步队列，以向生产者提供提要，并从消费者那里收集输出，以及如何干净地关闭它们(注意插入到队列中的哨兵)。此模式允许根据最近的输出生成新的输入，允许进行递归发现和处理。

3  

Nosklo's offered solution will quickly break if too much data is written to the receiving end of the pipe:

如果将过多的数据写入到管道的接收端，Nosklo提供的解决方案将会很快中断:

from subprocess import Popen, PIPE

p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE, close_fds=True)
p1.stdin.write('Hello World\n' * 20000)
p1.stdin.close()
result = p2.stdout.read() 
assert result == "Hello Worl\n"

If this script doesn't hang on your machine, just increase "20000" to something that exceeds the size of your operating system's pipe buffers.

如果此脚本不挂在您的机器上，只需将“20000”增加到超过操作系统的管道缓冲区的大小。

This is because the operating system is buffering the input to "grep", but once that buffer is full, the p1.stdin.write call will block until something reads from p2.stdout. In toy scenarios, you can get way with writing to/reading from a pipe in the same process, but in normal usage, it is necessary to write from one thread/process and read from a separate thread/process. This is true for subprocess.popen, os.pipe, os.popen*, etc.

这是因为操作系统正在将输入缓冲到“grep”，但是一旦缓冲区满了，p1.stdin就会出现。写调用将阻塞，直到从p2.stdout读取。在toy场景中，您可以通过在同一个进程中从管道中写入/读取的方式获得方法，但是在正常使用情况下，必须从一个线程/进程写入，并从一个单独的线程/进程中读取。这对于子流程来说是正确的。popen,操作系统。管、操作系统。popen *等。

Another twist is that sometimes you want to keep feeding the pipe with items generated from earlier output of the same pipe. The solution is to make both the pipe feeder and the pipe reader asynchronous to the man program, and implement two queues: one between the main program and the pipe feeder and one between the main program and the pipe reader. PythonInteract is an example of that.

另一个难题是，有时您需要继续为管道提供与以前相同管道的输出所生成的项。解决方案是让管道馈线和管道阅读器异步到man程序，并实现两个队列:一个在主程序和管道馈线之间，另一个在主程序和管道阅读器之间。python互动就是一个例子。

Subprocess is a nice convenience model, but because it hides the details of the os.popen and os.fork calls it does under the hood, it can sometimes be more difficult to deal with than the lower-level calls it utilizes. For this reason, subprocess is not a good way to learn about how inter-process pipes really work.

子过程是一个很好的便利模型，但是因为它隐藏了操作系统的细节。popen和操作系统。fork调用它在引擎盖下做，它有时比它使用的低级调用更难处理。因此，子过程不是了解进程间管道真正工作的好方法。

2  

You must do this in several threads. Otherwise, you'll end up in a situation where you can't send data: child p1 won't read your input since p2 doesn't read p1's output because you don't read p2's output.

您必须在几个线程中执行此操作。否则，你就会陷入无法发送数据的情况:p1不会读取你的输入，因为p2不会读取p1的输出，因为你不会读取p2的输出。

So you need a background thread that reads what p2 writes out. That will allow p2 to continue after writing some data to the pipe, so it can read the next line of input from p1 which again allows p1 to process the data which you send to it.

你需要一个后台线程来读取p2写出来的内容。这将允许p2在写入一些数据到管道之后继续，这样它就可以读取p1的下一行，这再次允许p1处理你发送给它的数据。

Alternatively, you can send the data to p1 with a background thread and read the output from p2 in the main thread. But either side must be a thread.

或者，您可以使用后台线程将数据发送到p1，并在主线程中读取p2的输出。但任何一方都必须是一条线。

2  

Responding to nosklo's assertion (see other comments to this question) that it can't be done without close_fds=True:

响应nosklo的断言(参见其他对这个问题的评论)，如果没有close_fds=True:

close_fds=True is only necessary if you've left other file descriptors open. When opening multiple child processes, it's always good to keep track of open files that might get inherited, and to explicitly close any that aren't needed:

如果您已经打开了其他的文件描述符，那么close_fds=True是必需的。在打开多个子进程时，最好保持对可能被继承的打开文件的跟踪，并明确关闭不需要的任何文件:

from subprocess import Popen, PIPE

p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p1.stdin.write('Hello World\n')
p1.stdin.close()
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
result = p2.stdout.read() 
assert result == "Hello Worl\n"

close_fds defaults to False because subprocess prefers to trust the calling program to know what it's doing with open file descriptors, and just provide the caller with an easy option to close them all if that's what it wants to do.

close_fds默认为False，因为subprocess更喜欢信任调用程序，以了解它使用打开的文件描述符做什么，并且只向调用者提供一个简单的选项来关闭它们，如果这是它想要做的事情。

But the real issue is that pipe buffers will bite you for all but toy examples. As I have said in my other answers to this question, the rule of thumb is to not have your reader and your writer open in the same process/thread. Anyone who wants to use the subprocess module for two-way communication would be well-served to study os.pipe and os.fork, first. They're actually not that hard to use if you have a good example to look at.

但真正的问题是，管道缓冲区会咬你，除了玩具例子。正如我在回答这个问题的其他回答中所说的，经验法则是不要让你的读者和你的作者在同一个进程/线程中打开。任何想要使用子过程模块进行双向通信的人都可以很好地学习操作系统。管道和操作系统。叉,放在第一位。如果你有一个很好的例子，它们其实并不难用。

1  

I think you may be examining the wrong problem. Certainly as Aaron says if you try to be both a producer to the beginning of a pipeline, and a consumer of the end of the pipeline, it is easy to get into a deadlock situation. This is the problem that communicate() solves.

我想你可能正在研究错误的问题。当然，正如Aaron所说，如果你试图同时成为管道的生产者，同时也是管道末端的消费者，那么很容易陷入僵局。这就是通信()解决的问题。

communicate() isn't exactly correct for you since stdin and stdout are on different subprocess objects; but if you take a look at the implementation in subprocess.py you'll see that it does exactly what Aaron suggested.

由于stdin和stdout在不同的子进程对象上，所以通信()并不完全正确。但是，如果您查看子流程中的实现。你会发现它和Aaron的建议完全一致。

Once you see that communicate both reads and writes, you'll see that in your second try communicate() competes with p2 for the output of p1:

一旦您看到了通信的读写，您将看到在您的第二次尝试通信()与p2的输出竞争p1:

p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
# ...
p1.communicate('data\n')       # reads from p1.stdout, as does p2

I am running on win32, which definitely has different i/o and buffering characteristics, but this works for me:

我在win32上运行，它确实有不同的I /o和缓冲特性，但是这对我来说很有用:

p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
t = threading.Thread(target=get_output, args=(p2,)) 
t.start()
p1.stdin.write('hello world\n' * 100000)
p1.stdin.close()
t.join()

I tuned the input size to produce a deadlock when using a naive unthreaded p2.read()

我调整了输入的大小，以在使用幼稚的无线程p2.read()时产生死锁

You might also try buffering into a file, eg

您还可以尝试缓冲到一个文件中。

fd, _ = tempfile.mkstemp()
os.write(fd, 'hello world\r\n' * 100000)
os.lseek(fd, 0, os.SEEK_SET)
p1 = Popen(["grep", "-v", "not"], stdin=fd, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
print p2.stdout.read()

That also works for me without deadlocks.

这对我也适用，没有死锁。

1  

In one of the comments above, I challenged nosklo to either post some code to back up his assertions about select.select or to upvote my responses he had previously down-voted. He responded with the following code:

在上面的一个评论中，我向nosklo提出了挑战，要么发布一些代码来支持他关于select的断言。请选择或支持我的回答，他之前投票赞成。他的答复如下:

from subprocess import Popen, PIPE
import select

p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE, close_fds=True)

data_to_write = 100000 * 'hello world\n'
to_read = [p2.stdout]
to_write = [p1.stdin]
b = [] # create buffer
written = 0


while to_read or to_write:
    read_now, write_now, xlist = select.select(to_read, to_write, [])
    if read_now:
        data = p2.stdout.read(1024)
        if not data:
            p2.stdout.close()
            to_read = []
        else:
            b.append(data)

    if write_now:
        if written 

from subprocess import Popen, PIPE
import select
import time

p1 = Popen(["cat"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cat"], stdin=p1.stdout, stdout=PIPE, close_fds=True)

data_to_write = 'hello world\n'
to_read = [p2.stdout]
to_write = [p1.stdin]
b = [] # create buffer
written = 0


while to_read or to_write:
    time.sleep(1)
    read_now, write_now, xlist = select.select(to_read, to_write, [])
    if read_now:
        print 'I am reading now!'
        data = p2.stdout.read(1024)
        if not data:
            p1.stdout.close()
            to_read = []
        else:
            b.append(data)

    if write_now:
        print 'I am writing now!'
        if written 

write, write, close file, read -> success
write, read -> hang